package com.ruyuan.rapid.core.processor;

import com.lmax.disruptor.dsl.ProducerType;
import com.ruyuan.rapid.common.concurrent.queue.flusher.ParallelFlusher;
import com.ruyuan.rapid.common.enums.ResponseCode;
import com.ruyuan.rapid.core.RapidConfig;
import com.ruyuan.rapid.core.context.HttpRequestWrapper;
import com.ruyuan.rapid.core.helper.ResponseHelper;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyBatchEventProcessor implements NettyProcessor{

    private RapidConfig rapidConfig;

    private NettyCoreProcessor nettyCoreProcessor;

    private static final String namePrefix = "rapid-flusher";

    private ParallelFlusher<HttpRequestWrapper> parallelFlusher;

    public NettyBatchEventProcessor(RapidConfig rapidConfig,NettyCoreProcessor nettyCoreProcessor){
        this.rapidConfig = rapidConfig;
        this.nettyCoreProcessor = nettyCoreProcessor;

        ParallelFlusher.Builder<HttpRequestWrapper> builder = new ParallelFlusher.Builder<HttpRequestWrapper> ();
        builder.setBufferSize(rapidConfig.getBufferSize())
                .setThreads(rapidConfig.getProcessThread())
                .setProducerType(ProducerType.MULTI)
                .setNamePrefix(namePrefix)
                .setWaitStrategy(rapidConfig.getATureWaitStrategy());
        BatchEventProcessorListener batchEventProcessorListener = new BatchEventProcessorListener();
        builder.setEventListener(batchEventProcessorListener);

        this.parallelFlusher = builder.build();
    }

    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) {
            this.parallelFlusher.add(httpRequestWrapper);
    }

    @Override
    public void start() {
        this.nettyCoreProcessor.start();
        this.parallelFlusher.start();
    }

    @Override
    public void shutdown() {
        this.nettyCoreProcessor.shutdown();
        this.parallelFlusher.shutdown();
    }

    public class BatchEventProcessorListener implements ParallelFlusher.EventListener<HttpRequestWrapper>{

        @Override
        public void onEvent(HttpRequestWrapper event) throws Exception {
            nettyCoreProcessor.process(event);
        }

        @Override
        public void onException(Throwable ex, long sequence, HttpRequestWrapper event) {
            FullHttpRequest fullHttpRequest = event.getFullHttpRequest();
            ChannelHandlerContext ctx = event.getCtx();
            try {
                log.error("请求处理失败");
                FullHttpResponse httpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
                //区分keep-alive和没有keep-alive的区别
                if(!HttpUtil.isKeepAlive(fullHttpRequest)){
                    ctx.writeAndFlush(httpResponse).addListener(ChannelFutureListener.CLOSE);
                }else {
                    fullHttpRequest.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                    ctx.writeAndFlush(httpResponse);
                }
            }catch (Exception e){
                //ignore e
                log.error("请求回写失败");
            }

        }
    }
}
